package com.gxlevi.canal_kafak.util;


import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;

import java.util.Properties;

public class KafkaSender {
    private String topic;

    public KafkaSender(String topic) {
        super();
        this.topic = topic;
    }

    public static void sendMessage(String topic, String key, String data) {
        Producer<String, String> producer = createProducer();
        producer.send(new KeyedMessage<>(topic, key, data));
    }

    private static Producer<String, String> createProducer() {
        Properties properties = new Properties();
        properties.put("metadata.broker.list", GlobalConfigUtil.kafkaBootstrapServers);
        properties.put("zookeeper.connect", GlobalConfigUtil.kafkaZookeeperConnect);
        properties.put("serializer.class", StringEncoder.class.getName());

        return new Producer<String, String>(new ProducerConfig(properties));
    }
}
